package org.shuzhou.i_ha;

import java.io.IOException;
import java.time.LocalDate;
import java.time.Period;
import java.time.format.DateTimeFormatter;

import org.shuzhou.model.PersonalInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 用到的数据集为data/people.txt
 * 计算每个民族的男女生人数，得到的结果为（男生人数，女生人数）
 * 将此案例做checkpoint持久化
 */
public class CheckPersonStateful {
    public static void main(String[] args) throws Exception {

        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        // 开始配置checkpoint
        // 每 1000ms 开始一次 checkpoint
        env.enableCheckpointing(1000);

        // 高级选项：

        // 设置模式为精确一次 (这是默认值)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 也可以设置为至少一次，能够降低延迟
        // env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

        // 确认 checkpoints 之间的时间会进行 500 ms
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

        // Checkpoint 必须在一分钟内完成，否则就会被抛弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        // 允许两个连续的 checkpoint 错误
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);

        // 同一时间只允许一个 checkpoint 进行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        /**
         * 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
         * 
         * 官方文档描述：可以配置周期存储 checkpoint 到外部系统中。
         * Externalized checkpoints 将他们的元数据写到持久化存储上并且在 job 失败的时候不会被自动删除。 
         * 这种方式下，如果你的 job 失败，你将会有一个现有的 checkpoint 去恢复。
         * 更多的细节请看 Externalized checkpoints 的部署文档。
         * 
         * 部署参考文档：https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/ops/state/checkpoints/#externalized-checkpoints
         * 
         * 从保留的checkpoint中恢复作业：bin/flink run -s :checkpointMetaDataPath [:runArgs]
         */

        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        /**
         * Flink 提供了多种 state backends，它用于指定状态的存储方式和位置。
         * 
         * 简单来说默认checkpint存储在内存中，用户可以使用state backends将checkpoint存储到其它系统。
         * Flink 内置了以下这些开箱即用的 state backends ：
         * - HashMapStateBackend
         * - EmbeddedRocksDBStateBackend
         * 如果不设置，默认使用 HashMapStateBackend。
         * 
         * 由于 HashMapStateBackend 将数据以对象形式存储在堆中，因此重用这些对象数据是不安全的。
         * 
         * EmbeddedRocksDBStateBackend 将正在运行中的状态数据保存在 RocksDB 数据库中，
         * RocksDB 数据库默认将数据存储在 TaskManager 的数据目录。
         * EmbeddedRocksDBStateBackend 是目前唯一支持增量 CheckPoint 的 State Backend
         * 
         * 由于 RocksDB 是 Flink 默认分发包的一部分，所以如果你没在代码中使用 RocksDB，则不需要添加此依赖。
         * 而且可以在 flink-conf.yaml 文件中通过 state.backend.type 配置 State Backend
         * 
         * 在 flink-conf.yaml 可以通过键 state.backend.type 设置默认的 State Backend。
         * state.backend.type : jobmanager \rocksdb 
         * state.backend: hashmap
         * 
         * 代码中添加以下依赖
         * <dependency>
         *   <groupId>org.apache.flink</groupId>
         *   <artifactId>flink-statebackend-rocksdb</artifactId>
         *   <version>1.17.2</version>
         *   <scope>provided</scope>
         * </dependency>
         * 
         */

        // 设置state backends
        // env.setStateBackend(new HashMapStateBackend());

        /**
         * 获取数据，数据源为Kafka，避免从端口获取中文数据乱码
         * 数据格式
         * 王矗馨,女,汉族,411424195001010028,19500101,150977754190,kk4jlw850x@sina.com,甘肃省临夏回族自治州和政县CNR路884号
         */
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("192.168.56.151:9092,192.168.56.152:9092,192.168.56.153:9092")
                .setTopics("persion")
                .setGroupId("root")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> people = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        /**
         * 将数据使用map，一对一封装为PersonalInfo对象
         */
        SingleOutputStreamOperator<PersonalInfo> ds = people.map(new MapFunction<String, PersonalInfo>() {
            @Override
            public PersonalInfo map(String value) throws Exception {
                // System.out.println(value.toString());
                String encode = new String(value.getBytes("UTF-8"), "UTF-8");
                String[] fields = encode.split(",");
                PersonalInfo info = new PersonalInfo(fields[0], fields[1], fields[2], fields[3], fields[4], fields[5],
                        fields[6], fields[7]);
                return info;
            }
        });

        /**
         * 使用map，将数据转换为（PersonalInfo对象,Tuple2(0,0)）
         * 其中Tuple2(0,0)用于保存后续的计算结果，用于记录每个民族的男女生人数，最终结果为（男生人数，女生人数）
         * 然后使用keyBy对民族进行分区
         */
        KeyedStream<Tuple2<PersonalInfo, Tuple2<Integer, Integer>>, Object> keyedStream = ds
                .map(new MapFunction<PersonalInfo, Tuple2<PersonalInfo, Tuple2<Integer, Integer>>>() {

                    @Override
                    public Tuple2<PersonalInfo, Tuple2<Integer, Integer>> map(PersonalInfo value) throws Exception {
                        // 转换为（PersonalInfo对象,Tuple2(0,0)）
                        return Tuple2.of(value, Tuple2.of(0, 0));
                    }
                })
                .keyBy(value -> value.f0.getEthnicity());

        /**
         * reduce聚合案例：记录每个民族的男女总数，记录到一个元组中（男生人数，女生人数）
         * 使用有状态算子 keyed state
         */
        keyedStream.reduce(new RichReduceFunction<Tuple2<PersonalInfo, Tuple2<Integer, Integer>>>() {
            // sum用于记录状态，f0=男生总数，f1=女生总数
            private transient ValueState<Tuple2<Integer, Integer>> sum;

            @Override
            public Tuple2<PersonalInfo, Tuple2<Integer, Integer>> reduce(
                    Tuple2<PersonalInfo, Tuple2<Integer, Integer>> value1,
                    Tuple2<PersonalInfo, Tuple2<Integer, Integer>> value2) throws Exception {

                // 获取sum状态
                Tuple2<Integer, Integer> currentSum = sum.value();

                // 如果sum为空，则进行初始化
                if (currentSum == null) {
                    currentSum = Tuple2.of(0, 0);

                    // 将value1进行累加
                    if (value1.f0.getGender().equals("男")) {
                        currentSum.f0 += 1;
                    }

                    if (value1.f0.getGender().equals("女")) {
                        currentSum.f1 += 1;
                    }
                }

                // 累加value2的值
                if (!value2.equals(null)) {
                    // 将value2进行累加
                    if (value2.f0.getGender().equals("男")) {
                        currentSum.f0 += 1;
                    }

                    if (value2.f0.getGender().equals("女")) {
                        currentSum.f1 += 1;
                    }
                }

                sum.update(currentSum);

                // 将计算结果返回 (民族：人数,总年龄）
                return Tuple2.of(value1.f0, currentSum);
            }

            /*
             * 对记录状态的ValueState<Tuple2<Integer, Integer>>进行构造
             * 源码中推荐写法
             */
            @Override
            public void open(Configuration config) throws IOException {
                // 构造时不初始化默认值
                ValueStateDescriptor<Tuple2<Integer, Integer>> descriptor = new ValueStateDescriptor<>(
                        "sum", // the state name
                        TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
                        }));
                sum = getRuntimeContext().getState(descriptor);
                // 在下面的代码中，判断如果state没有初始化则赋为默认值
                if (sum == null) {
                    sum.update(Tuple2.of(0, 0));
                }
            }

        }).print();

        env.execute("Flink Operator Example");
    }

    // 计算年龄的函数
    public static int calculateAge(String dateOfBirth) {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
        LocalDate birthDate = LocalDate.parse(dateOfBirth, formatter);
        LocalDate currentDate = LocalDate.now();
        Period period = Period.between(birthDate, currentDate);
        return period.getYears();
    }
}
